package com.chinamobile.iot.handler.impl;

import com.alibaba.fastjson.JSON;
import com.chinamobile.iot.config.BrokerConfiguration;
import com.chinamobile.iot.config.MqttConfiguration;
import com.chinamobile.iot.handler.model.BrokerInfo;
import com.chinamobile.iot.handler.sub.NettyMqttClient;
import com.chinamobile.iot.mqtt.spi.impl.SubscriptionTree;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;

import static com.chinamobile.iot.handler.model.BrokerInfo.NORMAL;
import static com.chinamobile.iot.handler.sub.SendSubConst.CLIENT_ID_SEND_SUB;

/**
 * Created by szl on 2017/7/11.
 * callback class when the children of znode /brokers/ changes， the zkHelper will call the method of this object
 */
@Component
@ConditionalOnExpression("${mqtt.distribute}")
public class BrokerMonitorImpl implements BrokerMonitor {

    private static Logger logger = LoggerFactory.getLogger(BrokerMonitorImpl.class);

    @Autowired
    private BrokerConfiguration brokerConfiguration;

    @Autowired
    private MqttConfiguration mqttConfiguration;

    @Autowired(required = false)
    private NettyMqttClient nettyMqttClient;

    /**
     * the current online broker info
     */
    private ConcurrentHashMap<String, BrokerInfo> mapBrokerInfo = new ConcurrentHashMap<>();

    /**
     * the whole broker info that had startup
     */
    private ConcurrentHashMap<String, BrokerInfo> wholeBrokerInfo = new ConcurrentHashMap<>();

    private ZkHelper zkHelper;
    private AtomicBoolean manager = new AtomicBoolean(false);
    private String managerBroker;

    private SubscriptionTree<String> tree2;
    private StringRedisTemplate redisTemplate;

    public SubscriptionTree<String> getTree2() {
        return tree2;
    }

    public String getManagerBroker() {
        return managerBroker;
    }

    public void setManagerBroker(String managerBroker) {
        this.managerBroker = managerBroker;
        /**
         * if the manager is me
         */
        if (managerBroker.equals(brokerConfiguration.getName())){
            manager.set(true);
            String data = zkHelper.getManagerData();
            String[] brokers = data.split(";");
            for (int i = 0; i < brokers.length; i++){
                if (!brokers[i].equals(brokerConfiguration.getName())){
                    wholeBrokerInfo.put(brokers[i], new BrokerInfo());
                }
            }
            checkAndCleanRedis();
        }else{
            manager.set(false);
        }
    }

    public void setTree2(SubscriptionTree<String> tree) {
        this.tree2 = tree;
    }

    public StringRedisTemplate getRedisTemplate() {
        return redisTemplate;
    }

    public void setRedisTemplate(StringRedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    public ConcurrentHashMap<String, BrokerInfo> getMapBrokerInfo() {
        return mapBrokerInfo;
    }

    public void setMapBrokerInfo(ConcurrentHashMap<String, BrokerInfo> mapBrokerInfo) {
        this.mapBrokerInfo = mapBrokerInfo;
    }

    @Override
    public void onBrokerChange(List<String> brokers) {
        synchronized (this) {
            HashMap<String, BrokerInfo> tempMap = new HashMap<>();
            LinkedList<BrokerInfo> disconnectBrokers = new LinkedList<>();
            LinkedList<BrokerInfo> newBrokers = new LinkedList<>();

            logger.info("onBrokerChang(...) is called.");
            brokers.forEach((s) -> {
                logger.info("{}", s);
            });
            logger.info("broker list end");


            brokers.forEach((s) -> {
                BrokerInfo brokerInfo = JSON.parseObject(s, BrokerInfo.class);
                tempMap.put(brokerInfo.getName(), brokerInfo);
                if (!mapBrokerInfo.containsKey(brokerInfo.getName())) {
                    if (!brokerInfo.getName().equals(brokerConfiguration.getName())
                            && !newBrokers.contains(brokerInfo)) {
                        newBrokers.add(brokerInfo);
                    }
                }
            });

            // find the disconnect brokers
            mapBrokerInfo.forEach((s, info) -> {
                if (!tempMap.containsKey(s)) {
                    if (!disconnectBrokers.contains(info)) {
                        disconnectBrokers.add(info);
                    }
                }
            });

            // 记录新结点的状态
            newBrokers.forEach((s) -> {
                if (s.getStatus().equals(NORMAL) && !s.getName().equals(brokerConfiguration.getName())) {
                    mapBrokerInfo.put(s.getName(), s);
                }
            });

            // 记录
            disconnectBrokers.forEach((s) -> {
                mapBrokerInfo.remove(s.getName());
            });

            if (!newBrokers.isEmpty()) {
                onNewBroker(newBrokers);
            }
            if (!disconnectBrokers.isEmpty()) {
                onDisconnectBrokers(disconnectBrokers);
            }
        }
    }

    @Override
    public void setZkHelper(ZkHelper zkHelper) {
        this.zkHelper = zkHelper;
    }

    private void onNewBroker(List<BrokerInfo> brokerInfos) {
        // 把曾经出现的所有broker都记录下来
        brokerInfos.forEach((b) -> {
            wholeBrokerInfo.put(b.getName(), b);
        });

        brokerInfos.forEach((broker) -> {
            logger.info("new broker is started, name:{}, {}:{}", broker.getName(), broker.getIp(), broker.getPort());
        });

        if (mqttConfiguration.isSubscribe()) {
            if (nettyMqttClient != null) {
                brokerInfos.forEach((brokerInfo) -> {
                    if (!brokerInfo.getName().equals(brokerConfiguration.getName())) {
                        nettyMqttClient.connect(brokerInfo.getIp(), brokerInfo.getPort(), CLIENT_ID_SEND_SUB, null, null, true, 120);
                    }
                });
            }
        }
    }

    private void onDisconnectBrokers(LinkedList<BrokerInfo> brokerInfos) {
        // 判别是不是broker掉线了，如果是broker掉线了，直接在checkAndCleanRedis中进行清理处理，后续不再需要处理
        // 如果不是broker掉线了，则不进行清理

        if (manager.get() == false) {
            for (int i = 0; i < brokerInfos.size(); i++) {
                BrokerInfo broker = brokerInfos.get(i);
                logger.info("broker is down, name:{}, {}:{}", broker.getName(), broker.getIp(), broker.getPort());
                if (managerBroker != null && managerBroker.equals(broker.getName())) {
                    zkHelper.removeManager(managerBroker);
                    zkHelper.tryBecomeManager(brokerConfiguration.getName());
                    managerBroker = zkHelper.getManager();
                    if (managerBroker.equals(brokerConfiguration.getName())) {
                        manager.set(true);
                        checkAndCleanRedis();
                        return;
                    } else {
                        manager.set(false);
                    }
                }
            }
        }


        try {
            if (manager.get()) {

                // 1 创建临时结点
                brokerInfos.forEach((s) -> {
                    zkHelper.beginCleanBroker(s.getName());
                });

                if (mqttConfiguration.isSubscribe()) {
                    // 清理清理本地subscriptiontree
                    brokerInfos.forEach((s) -> {
                        tree2.removeSubscribe(s.getName());
                    });

                    // 清理redis
                    brokerInfos.forEach((s) -> {
                        redisTemplate.delete(s.getName());
                    });
                    // 断开推送连接
                    if (nettyMqttClient != null) {
                        brokerInfos.forEach((s) -> {
                            nettyMqttClient.close(s.getIp(), s.getPort());
                        });
                    }
                }
            } else {
                // 清理清理本地subscriptiontree
                brokerInfos.forEach((s) -> {
                    if (mqttConfiguration.isSubscribe()) {
                        tree2.removeSubscribe(s.getName());
                        if (nettyMqttClient != null) {
                            nettyMqttClient.close(s.getIp(), s.getPort());
                        }
                    }
                });
            }
        } catch (Exception e) {

        } finally {
            if(manager.get()){
                // 清理完毕
                brokerInfos.forEach((s) -> {
                    zkHelper.finishCleanBroker(s.getName());
                });
            }
        }
    }

    private void checkAndCleanRedis() {
        wholeBrokerInfo.forEach((s, b) -> {
            if (!mapBrokerInfo.containsKey(s)) {
                zkHelper.beginCleanBroker(s);
                // 防止redis异常导致没有调用finishCleanBroker
                try{
                    redisTemplate.delete(s);
                    tree2.removeSubscribe(s);
                }
                catch (Exception e){
                }
                zkHelper.finishCleanBroker(s);
            }
        });
        mapBrokerInfo.forEach((s, b) -> {
            if (!zkHelper.isBrokerValid(s)) {
                try{
                    redisTemplate.delete(s);
                    tree2.removeSubscribe(s);
                }
                catch (Exception e){
                }
                zkHelper.finishCleanBroker(s);
            }
        });
    }
}
